package org.budo.canal.runningmonitor.zookeeper;

import org.budo.canal.runningmonitor.ServerRunningMonitorFactory;
import org.budo.support.java.net.util.JavaNetUtil;
import org.budo.support.lang.util.ProcessUtil;
import org.budo.support.lang.util.StringUtil;

import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
@Getter
@Setter
public class ZookeeperServerRunningMonitorFactory implements ServerRunningMonitorFactory {
    private ZkClientx zkClientx;

    private String zookeeperUrl;

    @Override
    public ServerRunningMonitor newServerRunningMonitor(String destination) {
        ServerRunningMonitor serverRunningMonitor = new ServerRunningMonitor();

        ZkClientx _zkClientx_0 = this._zkClientx_0();
        if (null != _zkClientx_0) {
            serverRunningMonitor.setZkClient(_zkClientx_0);
        }

        Integer currentProcessId = ProcessUtil.getCurrentProcessId();
        String address = currentProcessId + "@" + JavaNetUtil.getHostInnerFirstCached();
        ServerRunningData serverData = new ServerRunningData(address);

        serverRunningMonitor.setServerData(serverData);
        log.info("#90 serverRunningMonitor = " + serverRunningMonitor);

        serverRunningMonitor.setDestination(destination);

        serverRunningMonitor.init();
        return serverRunningMonitor;
    }

    private ZkClientx _zkClientx_0() {
        if (null != this.getZkClientx()) {
            return this.getZkClientx();
        }

        String zookeeperUrl = this.getZookeeperUrl();
        if (StringUtil.isEmpty(zookeeperUrl)) {
            log.error("#52 init, zookeeperUrl=" + zookeeperUrl + ", this=" + this);
            return null;
        }

        if (!JavaNetUtil.isPortUsing(zookeeperUrl)) {
            throw new IllegalArgumentException("#57 not ok zk, zookeeperUrl=" + zookeeperUrl + ", this=" + this);
        }

        ZkClientx zkClientx = new ZkClientx(zookeeperUrl);

        log.info("#59 init zkClientx=" + zkClientx + ", zookeeperUrl=" + zookeeperUrl + ", this=" + this);

        this.setZkClientx(zkClientx);
        return this.getZkClientx();
    }
}